package polimi.reds.optimizer;

import java.io.Serializable;

import java.util.ArrayList;
import java.util.Map;
import java.util.logging.Logger;

import polimi.reds.Filter;
import polimi.reds.NodeDescriptor;
import polimi.reds.broker.overlay.Overlay;
import polimi.reds.broker.routing.Router;
import polimi.reds.broker.routing.SubscriptionTable;
import polimi.reds.optimizer.TrafficLock;

public class SwapStrategy extends Thread {
	
	protected Logger logger = Logger.getLogger("polimi.reds.optimizer.SwapThread");
	private AbstractOptimizerStrategy optimizer;
	private TrafficLock trafficLock;
	private final static int STATE_IDLE=0,STATE_N0_INIT=1,STATE_N0_WAITING_A=2,STATE_N0_WAITING_B=3,
							 			  STATE_N0_WAITING_N1=4,STATE_N0_WAITING_N2=5,
										  STATE_N1_INIT=11,STATE_N1_WAITING_N0=12,STATE_N1_FINALIZING=13,
								          STATE_N2_INIT=21,STATE_N2_WAITING_N0=22;
	//private static final long QUEUE_ACTIVITY_TIMEOUT = 1000;
	private int state=STATE_IDLE;
	private int sessionID=0;
	private boolean receivedAck=false;
	private int ackCode=0;
	private long sessionExpiration=0;
	private Overlay overlay;
	private NodeDescriptor n0, n1, n2, me, receivedAckSource=null;
	private int maxNeighbors;
	private SubscriptionTable table=null;
	private int pubsubEvents=0;
	private ArrayList<Filter> n0filters = null, n1filters = null;
	
	public SwapStrategy(AbstractOptimizerStrategy optimizer, TrafficLock trafficLock, int maxNeighbors, Router router) {
		this.optimizer=optimizer;
		this.trafficLock=trafficLock;
		this.overlay = optimizer.getOverlay();
		this.me = overlay.getNodeDescriptor();
		this.maxNeighbors=maxNeighbors;
		this.table = router.getSubscriptionTable();
	}
	
	public synchronized void beginSwapSession(NodeDescriptor n0, NodeDescriptor n1, NodeDescriptor n2, int sessionID) {
		
		if (!this.isAlive()) {
			this.start();
		}
		
		logger.fine("Beginning swap on node: " + me.getID() + " (n0=" + n0.getID() + " n1=" + n1.getID() + " n2=" + n2.getID() + ") id=" + sessionID);
		
		if (state!=STATE_IDLE) {
			logger.fine(me.getID() + " swap cannot be initialized because node is in state: " + state + " ("+sessionID+")");
			if (me.equals(n0)) {
				// already in use
			} else {
				try {
					send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {-1,new Integer(sessionID)}, n0);
				} catch (Exception e) { 
					e.printStackTrace();
				}
			}
			return;
		}
		
		if (me.equals(n0) && !(overlay.isNeighborOf(n1) && overlay.isNeighborOf(n2))) {
			logger.fine(me.getID() + " swap cannot be initialized because of bad neighborhood: " + state + " (id=" + sessionID+")");
			return;
		}
		
		if (trafficLock.isLocked()) {
			logger.fine(me.getID() + " swap cannot be initialized because node is already locked: " + state + " (id=" + sessionID+")");
			return;
		}
		
		if (state==STATE_IDLE && me.equals(n2) && overlay.getNumberOfBrokers()>=maxNeighbors) {
			logger.fine(me.getID() + " swap cannot be initialized because node exceeds the maximum number of neighbors" + " (id=" + sessionID+")");
			try {
				send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {-1,new Integer(sessionID)}, n0);
			} catch (Exception e) { }
			return;
		}
		
		//if (state==STATE_IDLE && (System.currentTimeMillis() - trafficLock.getLastReportedQueueActivity())<QUEUE_ACTIVITY_TIMEOUT) {
		//	logger.severe(me.getID() + " swap cannot be initialized because node is using queues");
		//	try {
		//		send(Optimizer.REMOTE_ACK, new Serializable[] {new Boolean(true),new Integer(sessionID)}, n0);
		//	} catch (Exception e) { }
		//	return;
		//}
		
		this.n0=n0;
		this.n1=n1;
		this.n2=n2;
		this.sessionID=sessionID;
		this.sessionExpiration=System.currentTimeMillis()+1000000;
		if (me.equals(n0)) {
			setSwapState(STATE_N0_INIT);
		} else if (me.equals(n1)) {
			setSwapState(STATE_N1_INIT);
		} else if (me.equals(n2)) {
			setSwapState(STATE_N2_INIT);
		}
	}
	
	private synchronized void setSwapState(int newState) {
		logger.fine(me.getID() + " changing state from " + state + " to " + newState + " (" + sessionID + ")");
		optimizer.addToTrace("STATE FROM " + state + " TO " + newState);
		optimizer.addToTrace("MyTable is now: " + table);
		//receivedAck=false;
		state=newState;
	}
	
	public Integer getSwapState() {
		return state;
	}
	
	public synchronized void notifyAck(int ackCode, int sessionID, NodeDescriptor source, Serializable parameter) {
		
		while (receivedAck) {
			try {
				Thread.sleep(10);
			} catch (Exception e) {};
		}
			
		if (state!=STATE_IDLE && sessionID==this.sessionID) {
			logger.fine(me.getID() + " received good ack from " +source+ " (code=" + ackCode + ", session=" + sessionID + ", state="+state+")");
			this.ackCode=ackCode;
			receivedAckSource=source;
			receivedAck=true;
			
			if (ackCode>=0 && source.equals(n0) && me.equals(n1) && parameter instanceof ArrayList<?>) {
				n0filters = (ArrayList<Filter>) parameter;
				n1filters = (ArrayList<Filter>) table.getAllFiltersExcept(false, n0);
			} else if (ackCode>=0 && me.equals(n2) && source.equals(n1) && parameter instanceof ArrayList<?>) {
				ArrayList<ArrayList<Filter>> filters = (ArrayList<ArrayList<Filter>>) parameter;
				table.removeAllSubscriptions(n0);
				for (Filter f:filters.get(0)) {
					table.addSubscription(n0, f);
					optimizer.addToTrace(me + " ADDED SUBSCRIPTION: " + f + " (for " + n0 + ")");
					logger.fine(me + "ADDED SUBSCRIPTION: " + f + " (for " + n0 + ")");
				}
					
				table.removeAllSubscriptions(n1);
				for (Filter f:filters.get(1)) {
					table.addSubscription(n1, f);
					optimizer.addToTrace(me + " ADDED SUBSCRIPTION: " + f + " (for " + n1 + ")");
					logger.fine(me + " ADDED SUBSCRIPTION: " + f + " (for " + n1 + ")");
				}
			}
		} else if (state!=STATE_IDLE) {
			logger.fine(me.getID() + " received *BAD* ack from " + source + " (code=" + ackCode + ", session=" + sessionID + ", state=" + state + ")");
		} else {
			logger.fine(me.getID() + " received *BAD* ack from " +source+" (code=" + ackCode + ", session=" + sessionID + "!=" + this.sessionID + ", state="+state+")");
		}
	}
	
	public void notifyPubSubEvent() {
		pubsubEvents++;
	}
	
	public void run() {
		logger.fine("Starting SwapThread on node: " + me.getID());
		while (overlay.isRunning()) {
			try {
				Thread.sleep(100);
			} catch(InterruptedException e){}
			//Thread.yield();
			
			//synchronized(router) {
			while (state != STATE_IDLE) {
				try {
					sleep(100);
				} catch(InterruptedException e) {
					if (sessionExpiration<System.currentTimeMillis() && state!=STATE_IDLE) {
						logger.fine(me.getID() + " timeout while in state " + state + ". Aborting."+ " (" +  sessionID + ")");
						trafficLock.unlock();
						setSwapState(STATE_IDLE);
					}
					
					if (state==STATE_IDLE) {
						continue;
					}
				}
				
				switch (state) {
				case STATE_N0_INIT:
					synchronized(trafficLock) {
						if (!trafficLock.isLocked()) {
							trafficLock.lock(n0,n1,n2);
						} else {
							setSwapState(STATE_IDLE);
							continue;
						}
					}
					
					receivedAck=false;
					logger.fine(me.getID() + " locked! State: " + state + "."+ " (" +  sessionID + ")");
					optimizer.addStat(OptimizerStrategy.STAT_ATTEMPTED_RECONFIGURATIONS_COUNT,1.0);
					try {
						send(OptimizerStrategy.REMOTE_SWAP_N2, new Serializable[] {n1,new Integer(sessionID)}, n2);
					} catch (Exception e) {
						logger.fine(me.getID() + " error sending " + OptimizerStrategy.REMOTE_SWAP_N2 + " request to: " + n2.getID()+ " (" +  sessionID + ")");
						setSwapState(STATE_IDLE);
						trafficLock.unlock();
						continue;
					}

					setSwapState(STATE_N0_WAITING_A);
					continue;
				case STATE_N0_WAITING_A:
					if (!trafficLock.isLocked()) {
						logger.severe(me.getID() + "***BAD*** LOCKING STATUS IN STATE: " + state+ " (" +  sessionID + ")");
					}
					if (receivedAck) {
						receivedAck=false;
						if (ackCode<0) {
							logger.severe(me.getID() + " SWAP ABORTED BY " + receivedAckSource+ " (" +  sessionID + ")");
							try {
								if (receivedAckSource.equals(n2)) {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n1);
								} else {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n2);
								}
							} catch (Exception ee) { }
							setSwapState(STATE_IDLE);
							trafficLock.unlock();
						} else {
							try {
								send(OptimizerStrategy.REMOTE_SWAP_N1, new Serializable[] {n2,new Integer(sessionID)}, n1);
							} catch (Exception e) {
								logger.severe(me.getID() + " error sending " + OptimizerStrategy.REMOTE_SWAP_N1 + " request to: " + n1.getID()+ " (" +  sessionID + ")");
								try {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n2);
								} catch (Exception ee) { }
								setSwapState(STATE_IDLE);
								trafficLock.unlock();
								continue;
							}
							setSwapState(STATE_N0_WAITING_B);
						}
					}
					continue;
				case STATE_N0_WAITING_B:
					if (!trafficLock.isLocked()) {
						logger.severe(me.getID() + "***BAD*** LOCKING STATUS IN STATE: " + state+ " (" +  sessionID + ")");
					}
					if (receivedAck) {
						receivedAck=false;
						if (ackCode<0) {
							logger.fine(me.getID() + " SWAP ABORTED BY " + receivedAckSource+ " (" +  sessionID + ")");
							try {
								if (receivedAckSource.equals(n2)) {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n1);
								} else {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n2);
								}
							} catch (Exception ee) { }	
							setSwapState(STATE_IDLE);
							trafficLock.unlock();
						} else {
							try {
								ArrayList<Filter> filters = new ArrayList<Filter>(table.getAllFiltersExcept(false, n1));
								for (NodeDescriptor n:overlay.getNeighbors()) {
									if (n.isBroker() && !n.equals(n1) && !n.equals(n2)) {
										for (Object o:table.getAllFilters(n)) {
											Filter f = (Filter) o;
											if (!filters.contains(f)) {
												filters.add(f);
											}
										}	
									}
								}
								
								send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(0),new Integer(sessionID),filters}, n1);
								pubsubEvents=0;
								setSwapState(STATE_N0_WAITING_N1);
							} catch (Exception e) {
								try {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n1);
								} catch (Exception ee) { }	
								try {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n2);
								} catch (Exception ee) { }
								setSwapState(STATE_IDLE);
								trafficLock.unlock();
							}
						}
					}
					continue;
				case STATE_N0_WAITING_N1:
					if (!trafficLock.isLocked()) {
						logger.severe(me.getID() + "***BAD*** LOCKING STATUS IN STATE: " + state+ " (" +  sessionID + ")");
					}
					if (receivedAck) {
						receivedAck=false;
						if (ackCode<0) {
							logger.fine(me.getID() + " SWAP ABORTED BY " + receivedAckSource+ " (" +  sessionID + ")");
							try {
								if (receivedAckSource.equals(n2)) {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n1);
								} else {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n2);
								}
							} catch (Exception ee) { }
							setSwapState(STATE_IDLE);
							trafficLock.unlock();
						} else if (ackCode==Integer.MAX_VALUE) {
							logger.fine(me.getID() + " SWAP ***SUCCESSFUL*** AS N0 (" + sessionID + ")");
							setSwapState(STATE_IDLE);
							trafficLock.unlock();
						} else {
							try {
								if (pubsubEvents==0) {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {ackCode+1,new Integer(sessionID)}, n2);
								} else {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {0,new Integer(sessionID)}, n2);
									pubsubEvents=0;
								}
							} catch (Exception ee) { }
							setSwapState(STATE_N0_WAITING_N2);
						}
					}
					continue;
				case STATE_N0_WAITING_N2:
					if (!trafficLock.isLocked()) {
						logger.severe(me.getID() + "***BAD*** LOCKING STATUS IN STATE: " + state+ " (" +  sessionID + ")");
					}
					if (receivedAck) {
						receivedAck=false;
						if (ackCode<0) {
							logger.fine(me.getID() + " SWAP ABORTED BY " + receivedAckSource+ " (" +  sessionID + ")");
							try {
								if (receivedAckSource.equals(n2)) {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n1);
								} else {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n2);
								}
							} catch (Exception ee) { }
							setSwapState(STATE_IDLE);
							trafficLock.unlock();
					    } else {
							try {
								if (pubsubEvents==0) {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {ackCode+1,new Integer(sessionID)}, n1);
								} else {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {0,new Integer(sessionID)}, n1);
									pubsubEvents=0;
								}
							} catch (Exception ee) { }
							setSwapState(STATE_N0_WAITING_N1);
						}
					}
					continue;
				case STATE_N1_INIT:
					synchronized(trafficLock) {
						if (!trafficLock.isLocked() && !overlay.isNeighborOf(n2)) {
							trafficLock.lock(n0,n1,n2);
						} else {
							try {
								send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n0);
							} catch (Exception e) { }
							setSwapState(STATE_IDLE);
							continue;
						}
					}
					
					try {
						logger.fine(me.getID() + " locked! State: " + state + "."+ " (" +  sessionID + ")");
						receivedAck=false;
						send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(0),new Integer(sessionID)}, n0);
						pubsubEvents=0;
						setSwapState(STATE_N1_WAITING_N0);
					} catch (Exception e) {
						logger.severe(me.getID() + " error sending ack back to N0"+ " (" +  sessionID + ")");
						trafficLock.unlock();
						setSwapState(STATE_IDLE);
					}
					continue;
				case STATE_N1_WAITING_N0:
					if (!trafficLock.isLocked()) {
						logger.severe(me.getID() + "***BAD*** LOCKING STATUS IN STATE: " + state+ " (" +  sessionID + ")");
					}
					if (receivedAck) {
						receivedAck=false;
						if (ackCode<0) {
							logger.fine(me.getID() + " SWAP ABORTED BY " + receivedAckSource+ " (" +  sessionID + ")");
							setSwapState(STATE_IDLE);
							trafficLock.unlock();
						} else if (ackCode>7) {
							logger.fine(me.getID() + " FINALIZING SWAP ");
							try {
								ArrayList<ArrayList<Filter>> filters = new ArrayList<ArrayList<Filter>>();
								filters.add(n0filters);
								filters.add(n1filters);
								overlay.removeNeighbor(n0);
								overlay.addNeighbor(n2.getUrls()[0]);
								logger.severe(me.getID() + " swap performed, sending ack..."+ " (" +  sessionID + ")");
								send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {Integer.MAX_VALUE,new Integer(sessionID),filters}, n2);
								setSwapState(STATE_N1_FINALIZING);				
							} catch (Exception e) {
								logger.severe(me.getID() + " ***SERIOUS*** error while performing swap (state: " + state + ")");
								setSwapState(STATE_IDLE);
								trafficLock.unlock();
							}
						} else {
							receivedAck=false;
							try {
								if (pubsubEvents==0) {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {ackCode+1,new Integer(sessionID)}, n0);
								} else {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {0,new Integer(sessionID)}, n0);
									pubsubEvents=0;
								}
							} catch (Exception ee) { }
						}
					}
					continue;
				case STATE_N1_FINALIZING:
					if (!trafficLock.isLocked()) {
						logger.severe(me.getID() + "***BAD*** LOCKING STATUS IN STATE: " + state+ " (" +  sessionID + ")");
					} 
					if (receivedAck) {
						receivedAck=false;
						if (ackCode<0) {
							logger.fine(me.getID() + " SWAP ABORTED BY " + receivedAckSource+ " (" +  sessionID + ")");
						} else {
							logger.fine(me.getID() + " SWAP ***SUCCESSFUL*** AS N1 (" + sessionID + ")");	
							optimizer.addStat(OptimizerStrategy.STAT_SUCCESSFUL_RECONFIGURATIONS_COUNT,1.0);
						}
						setSwapState(STATE_IDLE);
						trafficLock.unlock();
					}
					continue;
				case STATE_N2_INIT:
					synchronized(trafficLock) {
						if (!trafficLock.isLocked() && !overlay.isNeighborOf(n1)) {
							trafficLock.lock(n0,n1,n2);
						} else {
							try {
								send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(-1),new Integer(sessionID)}, n0);
							} catch (Exception e) { }
							setSwapState(STATE_IDLE);
							continue;
						}
					}			
					receivedAck=false;
					
					try {
						logger.fine(me.getID() + " locked! State: " + state + "." + " (" +  sessionID + ")");
						send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {new Integer(0),new Integer(sessionID)}, n0);
						pubsubEvents=0;
						setSwapState(STATE_N2_WAITING_N0);
					} catch (Exception e) {
						logger.severe(me.getID() + " error sending ack back to N0 "+ " (" +  sessionID + ")");
						setSwapState(STATE_IDLE);
						trafficLock.unlock();
					}
					continue;
				case STATE_N2_WAITING_N0:
					if (!trafficLock.isLocked()) {
						logger.severe(me.getID() + "***BAD*** LOCKING STATUS IN STATE: " + state+ " (" +  sessionID + ")");
					} 
					if (receivedAck) {
						receivedAck=false;
						if (ackCode<0) {
							logger.severe(me.getID() + " SWAP ABORTED BY " + receivedAckSource + " (" +  sessionID + ")");
							setSwapState(STATE_IDLE);
							trafficLock.unlock();
						} else if (ackCode==Integer.MAX_VALUE) {
							logger.fine(me.getID() + " finalizing swap ");
							try {
								logger.fine(me.getID() + " swap performed, sending ack...");
								send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {Integer.MAX_VALUE,new Integer(sessionID)}, n1);
								send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {Integer.MAX_VALUE,new Integer(sessionID)}, n0);
								logger.fine(me.getID() + " SWAP ***SUCCESSFUL*** AS N1 (" + sessionID + ")");	
							} catch (Exception e) {
								logger.severe(me.getID() + " ***SERIOUS*** error while sending ack (state: " + state + ") " + e.getMessage()+ " (" +  sessionID + ")");
							}
							setSwapState(STATE_IDLE);
							trafficLock.unlock();
						} else {
							try {
								if (pubsubEvents==0) {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {ackCode+1,new Integer(sessionID)}, n0);
								} else {
									send(OptimizerStrategy.REMOTE_ACK, new Serializable[] {0,new Integer(sessionID)}, n0);
									pubsubEvents=0;
								}
							} catch (Exception ee) { }
						}
					}
					continue;
				}
			}
		}
		
		logger.severe("Stopping SwapThread on node: " + me.getID());
	}
	
	private synchronized void send(String subject, Serializable packet, NodeDescriptor recipient) throws Exception {
		//logger.severe(me.getID() + " SENDTO " + recipient.getID() + ": " + subject + " (" + p(packet) + ")");
		overlay.send(subject, packet, recipient);
		optimizer.addStat(OptimizerStrategy.STAT_RECONFIGURATION_TRAFFIC_EVENTS,1.0);
		optimizer.addStat(OptimizerStrategy.STAT_RECONFIGURATION_TRAFFIC_SIZE,new Double(optimizer.packetSize(subject, packet, recipient)));
	}
	
	public static String p(Serializable s) {
		String res="";
		
		if (s.getClass().isArray()) {
			Object s1[] = (Object[]) s;
			for (Object i:s1) {
				res += i.toString() + " ";
			}
			return res;
		}
		
		return s.toString();	
	}
}
